package com.we.risk.behavior.repay;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;

import static com.we.risk.behavior.repay.CommonVar.*;

public class TableLoan {
    public static final int HHCODE = 1;
    public static final String KFKTABLE = "fdm.uni01_loan";
    public static final String QUERYSQL =
            "select loan_key, "
                    + "channel_code, update_time "
                    + "from "
                    + KFKTABLE
                    + " where channel_code = 'HAOHUAN' " +
                    "and update_time is not null " +
                    "and loan_key is not null";

    public static Logger LOG = LoggerFactory.getLogger(TableLoan.class);

    public String loan_key;
    public String channel_code;
    public Timestamp update_time;

    public TableLoan() {
    }

    public static TableLoan init(String loan_key) {
        return new TableLoan(loan_key, null, null);
    }

    public TableLoan(String loan_key, String channel_code, Timestamp update_time) {
        this.loan_key = loan_key;
        this.channel_code = channel_code;
        this.update_time = update_time;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableLoan.QUERYSQL);
        DataStream<Tuple2<Boolean, TableLoan>> inputStream =
                tableEnv.toRetractStream(tableRes, TableLoan.class);

        SingleOutputStreamOperator<String> out = inputStream.filter(new FilterFunction<Tuple2<Boolean, TableLoan>>() {
            @Override
            public boolean filter(Tuple2<Boolean, TableLoan> value) throws Exception {
                return value.f0;
            }
        }).uid("TableLoan-filter")
        .keyBy(new KeySelector<Tuple2<Boolean, TableLoan>, String>() {
            @Override
            public String getKey(Tuple2<Boolean, TableLoan> value) throws Exception {
                return value.f1.getLoan_key();
            }
        }).flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableLoan>, String>() {
            ValueState<TableLoan> tableLoanVState;

            @Override
            public void open(Configuration parameters) throws Exception {
                tableLoanVState = getRuntimeContext()
                        .getState(new ValueStateDescriptor<TableLoan>("TableLoanVState", TableLoan.class));
            }

            @Override
            public void flatMap(Tuple2<Boolean, TableLoan> value, Collector<String> out) throws Exception {
                try {
                    String loan_key = value.f1.getLoan_key();
                    String channel_code = value.f1.getChannel_code();
                    Timestamp update_time_stamp = value.f1.getUpdate_time();
                    long update_time = update_time_stamp.getTime();

                    TableLoan tmpState = tableLoanVState.value();
                    TableLoan curState = tmpState == null ? init(loan_key) : tmpState;

                    Timestamp state_time = curState.getUpdate_time();
                    if (state_time == null || update_time > state_time.getTime()) {
                        String state_channel_code = curState.getChannel_code();
                        if (state_channel_code == null || !state_channel_code.equals(channel_code)) {
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(LOANKEY, loan_key);
                            jsonObject.put(CHANNELCODE, HHCODE); //压缩kfk数据

                            out.collect(jsonObject.toString());

                            curState.channel_code = channel_code;
                            curState.update_time = update_time_stamp;
                            curState.loan_key = loan_key;
                        }
                    }
                } catch (Exception e) {
                    LOG.error("TableLoan flatMap exec: " + e.toString());
                    LOG.error("TableLoan flatMap exec input: " + value.toString());
                }

            }
        });
        return out;
    }

    public String getLoan_key() {
        return loan_key;
    }

    public String getChannel_code() {
        return channel_code;
    }

    public Timestamp getUpdate_time() {
        return update_time;
    }
}
